From 4bc3d82773185e4dfb54fcbd935f04c3ea0f0e59 Mon Sep 17 00:00:00 2001 From: grabowski Date: Wed, 22 Jul 2026 12:51:29 +0700 Subject: [PATCH] Persist station CRUD across restarts via JSON config Station CRUD via the API previously mutated the scraper's in-memory station_mapping only, so changes were lost on restart (and the systemd service auto-restarts). - Extract the 130-line hardcoded station_mapping into bundled defaults at src/data/stations.json; the scraper loads from a runtime-writable config file (STATION_CONFIG_PATH, default stations.json) and falls back to the bundled defaults to seed it. - Add scraper.save_stations() with an atomic temp-file + os.replace write. - create/update/delete station endpoints now persist and roll back the in-memory change if the write fails; re-raise HTTPException so persistence errors surface as real 500s instead of being swallowed. - Backend-agnostic (works for the VictoriaMetrics deployment, which has no relational stations table). Runtime stations.json is gitignored. Also clears pre-existing flake8 debt in water_scraper_v3.py (unused imports, long lines, duplicate logging import) and dedupes the User-Agent to Config.USER_AGENT. --- .gitignore | 4 +- src/config.py | 6 + src/data/stations.json | 130 ++++++++++++ src/water_scraper_v3.py | 459 ++++++++++++++++++---------------------- src/web_api.py | 24 ++- 5 files changed, 362 insertions(+), 261 deletions(-) create mode 100644 src/data/stations.json diff --git a/.gitignore b/.gitignore index 599d35b..22b9916 100644 --- a/.gitignore +++ b/.gitignore @@ -134,4 +134,6 @@ cython_debug/ # Docker volumes vm_data/ -grafana_data/ \ No newline at end of file +grafana_data/ +# Runtime station config (persisted CRUD); bundled default lives in src/data/ +/stations.json diff --git a/src/config.py b/src/config.py index 8b8bd2c..d7aed3a 100644 --- a/src/config.py +++ b/src/config.py @@ -88,6 +88,12 @@ class Config: MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3")) RETRY_DELAY_SECONDS = int(os.getenv("RETRY_DELAY_SECONDS", "60")) + # Station configuration + # Runtime-writable JSON file that persists the station mapping across restarts + # (station CRUD via the API writes here). If it does not exist, the bundled + # defaults in src/data/stations.json are used to seed it. + STATION_CONFIG_PATH = os.getenv("STATION_CONFIG_PATH", "stations.json") + # Web API / CORS settings # Comma-separated list of allowed origins. Defaults to none (same-origin only); # set CORS_ALLOW_ORIGINS to a specific list of front-end origins in production. diff --git a/src/data/stations.json b/src/data/stations.json new file mode 100644 index 0000000..c849b1c --- /dev/null +++ b/src/data/stations.json @@ -0,0 +1,130 @@ +{ + "1": { + "code": "P.20", + "thai_name": "บ้านเชียงดาว", + "english_name": "Ban Chiang Dao", + "latitude": 19.36731448032191, + "longitude": 98.9688487015384, + "geohash": null + }, + "2": { + "code": "P.75", + "thai_name": "บ้านช่อแล", + "english_name": "Ban Chai Lat", + "latitude": 19.145972935976225, + "longitude": 99.00735727149247, + "geohash": null + }, + "3": { + "code": "P.92", + "thai_name": "บ้านเมืองกึ๊ด", + "english_name": "Ban Muang Aut", + "latitude": 19.220518985435646, + "longitude": 98.84733127007874, + "geohash": null + }, + "4": { + "code": "P.4A", + "thai_name": "บ้านแม่แตง", + "english_name": "Ban Mae Taeng", + "latitude": 19.1222679952378, + "longitude": 98.94437462084075, + "geohash": null + }, + "5": { + "code": "P.67", + "thai_name": "บ้านแม่แต", + "english_name": "Ban Tae", + "latitude": 19.009762080002453, + "longitude": 98.95978297135508, + "geohash": null + }, + "6": { + "code": "P.21", + "thai_name": "บ้านริมใต้", + "english_name": "Ban Rim Tai", + "latitude": 18.917459157963293, + "longitude": 98.97018092996231, + "geohash": null + }, + "7": { + "code": "P.103", + "thai_name": "สะพานวงแหวนรอบ 3", + "english_name": "Ring Bridge 3", + "latitude": 18.86664807441675, + "longitude": 98.9781107622432, + "geohash": null + }, + "8": { + "code": "P.1", + "thai_name": "สะพานนวรัฐ", + "english_name": "Nawarat Bridge", + "latitude": 18.7875, + "longitude": 99.0045, + "geohash": "w5q6uuhvfcfp25" + }, + "9": { + "code": "P.82", + "thai_name": "บ้านสบวิน", + "english_name": "Ban Sob win", + "latitude": 18.6519444, + "longitude": 98.69, + "geohash": null + }, + "10": { + "code": "P.84", + "thai_name": "บ้านพันตน", + "english_name": "Ban Panton", + "latitude": 18.591315274591334, + "longitude": 98.79657058508496, + "geohash": null + }, + "11": { + "code": "P.81", + "thai_name": "บ้านโป่ง", + "english_name": "Ban Pong", + "latitude": 13.805661820610888, + "longitude": 99.87174946122846, + "geohash": null + }, + "12": { + "code": "P.5", + "thai_name": "สะพานท่านาง", + "english_name": "Tha Nang Bridge", + "latitude": 18.580269437546555, + "longitude": 99.01021397084362, + "geohash": null + }, + "13": { + "code": "P.77", + "thai_name": "บ้านสบแม่สะป๊วด", + "english_name": "Baan Sop Mae Sapuord", + "latitude": 18.433347475179602, + "longitude": 99.08510036666527, + "geohash": null + }, + "14": { + "code": "P.87", + "thai_name": "บ้านป่าซาง", + "english_name": "Ban Pa Sang", + "latitude": 18.519121825282486, + "longitude": 98.94224374138238, + "geohash": null + }, + "15": { + "code": "P.76", + "thai_name": "บ้านแม่อีไฮ", + "english_name": "Banb Mae I Hai", + "latitude": 18.141465831254404, + "longitude": 98.89642508267181, + "geohash": null + }, + "16": { + "code": "P.85", + "thai_name": "บ้านหล่ายแก้ว", + "english_name": "Baan Lai Kaew", + "latitude": 18.17856361002219, + "longitude": 98.63023114782287, + "geohash": null + } +} diff --git a/src/water_scraper_v3.py b/src/water_scraper_v3.py index 02c6426..97c410a 100644 --- a/src/water_scraper_v3.py +++ b/src/water_scraper_v3.py @@ -3,240 +3,171 @@ Enhanced Water Monitor Scraper with multiple database backend support """ -import requests import datetime -import time -import schedule import json +import logging import os -from typing import List, Dict, Optional +import time +from typing import Dict, List, Optional + +import requests +import schedule try: - from .database_adapters import create_database_adapter, DatabaseAdapter - from .models import WaterMeasurement, StationInfo, ScrapingResult, StationStatus - from .validators import DataValidator - from .exceptions import APIConnectionError, DataValidationError, DatabaseConnectionError - from .metrics import increment_counter, set_gauge, record_histogram, Timer - from .rate_limiter import RateLimiter, RequestTracker + from .config import Config + from .database_adapters import create_database_adapter from .logging_config import get_logger + from .metrics import Timer, increment_counter, record_histogram, set_gauge + from .rate_limiter import RateLimiter, RequestTracker + from .validators import DataValidator except ImportError: # Handle case when running as standalone script - from database_adapters import create_database_adapter, DatabaseAdapter - import logging - + from config import Config + from database_adapters import create_database_adapter + def get_logger(name): return logging.getLogger(name) - + def increment_counter(*args, **kwargs): pass - + def set_gauge(*args, **kwargs): pass - + def record_histogram(*args, **kwargs): pass - + class Timer: def __init__(self, *args, **kwargs): pass + def __enter__(self): return self + def __exit__(self, *args): pass - + class RateLimiter: def __init__(self, *args, **kwargs): pass + def wait_if_needed(self): pass - + class RequestTracker: def __init__(self): pass + def record_request(self, *args, **kwargs): pass - + class DataValidator: @staticmethod def validate_measurements(measurements): return measurements + # Get logger instance logger = get_logger(__name__) + class EnhancedWaterMonitorScraper: def __init__(self, db_config: Dict): """ Initialize scraper with database configuration - + Args: db_config: Database configuration dictionary """ self.api_url = "https://hyd-app-db.rid.go.th/webservice/getGroupHourlyWaterLevelReportAllHL.ashx" self.db_config = db_config.copy() # Make a copy to avoid modifying original self.db_adapter = None - + # Scheduler state tracking self.last_successful_update = None self.retry_mode = False self.next_hourly_check = None - + # Rate limiting and request tracking self.rate_limiter = RateLimiter(max_requests=10, time_window_seconds=60) self.request_tracker = RequestTracker() - + # HTTP session for API requests self.session = requests.Session() - self.session.headers.update({ - 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36', - 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', - 'Accept': 'application/json, text/javascript, */*; q=0.01', - 'X-Requested-With': 'XMLHttpRequest' - }) - - # Station mapping with correct names and geolocation data - self.station_mapping = { - '1': { - 'code': 'P.20', - 'thai_name': 'บ้านเชียงดาว', - 'english_name': 'Ban Chiang Dao', - 'latitude': 19.36731448032191, - 'longitude': 98.9688487015384, - 'geohash': None - }, - '2': { - 'code': 'P.75', - 'thai_name': 'บ้านช่อแล', - 'english_name': 'Ban Chai Lat', - 'latitude': 19.145972935976225, - 'longitude': 99.00735727149247, - 'geohash': None - }, - '3': { - 'code': 'P.92', - 'thai_name': 'บ้านเมืองกึ๊ด', - 'english_name': 'Ban Muang Aut', - 'latitude': 19.220518985435646, - 'longitude': 98.84733127007874, - 'geohash': None - }, - '4': { - 'code': 'P.4A', - 'thai_name': 'บ้านแม่แตง', - 'english_name': 'Ban Mae Taeng', - 'latitude': 19.1222679952378, - 'longitude': 98.94437462084075, - 'geohash': None - }, - '5': { - 'code': 'P.67', - 'thai_name': 'บ้านแม่แต', - 'english_name': 'Ban Tae', - 'latitude': 19.009762080002453, - 'longitude': 98.95978297135508, - 'geohash': None - }, - '6': { - 'code': 'P.21', - 'thai_name': 'บ้านริมใต้', - 'english_name': 'Ban Rim Tai', - 'latitude': 18.917459157963293, - 'longitude': 98.97018092996231, - 'geohash': None - }, - '7': { - 'code': 'P.103', - 'thai_name': 'สะพานวงแหวนรอบ 3', - 'english_name': 'Ring Bridge 3', - 'latitude': 18.86664807441675, - 'longitude': 98.9781107622432, - 'geohash': None - }, - '8': { - 'code': 'P.1', - 'thai_name': 'สะพานนวรัฐ', - 'english_name': 'Nawarat Bridge', - 'latitude': 18.7875, - 'longitude': 99.0045, - 'geohash': 'w5q6uuhvfcfp25' - }, - '9': { - 'code': 'P.82', - 'thai_name': 'บ้านสบวิน', - 'english_name': 'Ban Sob win', - 'latitude': 18.6519444, - 'longitude': 98.69, - 'geohash': None - }, - '10': { - 'code': 'P.84', - 'thai_name': 'บ้านพันตน', - 'english_name': 'Ban Panton', - 'latitude': 18.591315274591334, - 'longitude': 98.79657058508496, - 'geohash': None - }, - '11': { - 'code': 'P.81', - 'thai_name': 'บ้านโป่ง', - 'english_name': 'Ban Pong', - 'latitude': 13.805661820610888, - 'longitude': 99.87174946122846, - 'geohash': None - }, - '12': { - 'code': 'P.5', - 'thai_name': 'สะพานท่านาง', - 'english_name': 'Tha Nang Bridge', - 'latitude': 18.580269437546555, - 'longitude': 99.01021397084362, - 'geohash': None - }, - '13': { - 'code': 'P.77', - 'thai_name': 'บ้านสบแม่สะป๊วด', - 'english_name': 'Baan Sop Mae Sapuord', - 'latitude': 18.433347475179602, - 'longitude': 99.08510036666527, - 'geohash': None - }, - '14': { - 'code': 'P.87', - 'thai_name': 'บ้านป่าซาง', - 'english_name': 'Ban Pa Sang', - 'latitude': 18.519121825282486, - 'longitude': 98.94224374138238, - 'geohash': None - }, - '15': { - 'code': 'P.76', - 'thai_name': 'บ้านแม่อีไฮ', - 'english_name': 'Banb Mae I Hai', - 'latitude': 18.141465831254404, - 'longitude': 98.89642508267181, - 'geohash': None - }, - '16': { - 'code': 'P.85', - 'thai_name': 'บ้านหล่ายแก้ว', - 'english_name': 'Baan Lai Kaew', - 'latitude': 18.17856361002219, - 'longitude': 98.63023114782287, - 'geohash': None + self.session.headers.update( + { + "User-Agent": Config.USER_AGENT, + "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", + "Accept": "application/json, text/javascript, */*; q=0.01", + "X-Requested-With": "XMLHttpRequest", } - } - + ) + + # Station mapping is persisted to a JSON file so that station CRUD via the + # API survives restarts; on first run it is seeded from the bundled + # defaults in data/stations.json. + self.station_config_path = Config.STATION_CONFIG_PATH + self.station_mapping = self._load_station_mapping() + self.init_database() - + + @staticmethod + def _default_station_mapping_path() -> str: + """Path to the bundled default station mapping shipped with the package.""" + return os.path.join(os.path.dirname(os.path.abspath(__file__)), "data", "stations.json") + + def _load_station_mapping(self) -> Dict: + """Load the station mapping, preferring the runtime-writable config file. + + Order of precedence: + 1. The runtime config file (STATION_CONFIG_PATH) if it exists — this holds + any changes made through the station CRUD API. + 2. The bundled defaults in data/stations.json. + """ + for source in (self.station_config_path, self._default_station_mapping_path()): + if source and os.path.exists(source): + try: + with open(source, encoding="utf-8") as f: + mapping = json.load(f) + logger.info(f"Loaded {len(mapping)} stations from {source}") + return mapping + except Exception as e: + logger.error(f"Failed to load station mapping from {source}: {e}") + + logger.error("No station mapping could be loaded; starting with an empty mapping") + return {} + + def save_stations(self) -> bool: + """Persist the current station mapping to the runtime config file. + + Written atomically (temp file + replace) so a crash mid-write cannot + corrupt the existing configuration. + """ + path = self.station_config_path + if not path: + logger.warning("STATION_CONFIG_PATH not set; station changes will not persist") + return False + try: + tmp_path = f"{path}.tmp" + with open(tmp_path, "w", encoding="utf-8") as f: + json.dump(self.station_mapping, f, ensure_ascii=False, indent=2) + f.write("\n") + os.replace(tmp_path, path) + logger.info(f"Persisted {len(self.station_mapping)} stations to {path}") + return True + except Exception as e: + logger.error(f"Failed to persist station mapping to {path}: {e}") + return False + def init_database(self): """Initialize database connection""" try: # Extract db_type and pass remaining config as kwargs db_config_copy = self.db_config.copy() - db_type = db_config_copy.pop('type') + db_type = db_config_copy.pop("type") self.db_adapter = create_database_adapter(db_type, **db_config_copy) success = self.db_adapter.connect() - + if success: logger.info(f"Successfully connected to {db_type.upper()} database") set_gauge("database_connected", 1) @@ -245,53 +176,53 @@ class EnhancedWaterMonitorScraper: logger.error(f"Failed to connect to {db_type.upper()} database") set_gauge("database_connected", 0) increment_counter("database_connections_failed") - + except Exception as e: logger.error(f"Error initializing database: {e}") set_gauge("database_connected", 0) increment_counter("database_connections_failed") self.db_adapter = None - + def fetch_water_data_for_date(self, target_date: datetime.datetime) -> Optional[List[Dict]]: """Fetch water levels and discharge data from API for a specific date""" with Timer("api_request_duration"): try: logger.info(f"Starting data fetch from API for date: {target_date.strftime('%Y-%m-%d')}") - + # Rate limiting self.rate_limiter.wait_if_needed() - + # Create Thai format date (Buddhist calendar) thai_year = target_date.year + 543 thai_date = f"{target_date.day:02d}/{target_date.month:02d}/{thai_year}" - + # API parameters payload = { - 'DW[UtokID]': '1', - 'DW[BasinID]': '6', - 'DW[TimeCurrent]': thai_date, - '_search': 'false', - 'nd': str(int(time.time() * 1000)), - 'rows': '100', - 'page': '1', - 'sidx': 'indexhourly', - 'sord': 'asc' + "DW[UtokID]": "1", + "DW[BasinID]": "6", + "DW[TimeCurrent]": thai_date, + "_search": "false", + "nd": str(int(time.time() * 1000)), + "rows": "100", + "page": "1", + "sidx": "indexhourly", + "sord": "asc", } - + logger.debug(f"API parameters: {payload}") - + # POST request to API start_time = time.time() response = self.session.post(self.api_url, data=payload, timeout=30) response_time = time.time() - start_time - + response.raise_for_status() - + # Record successful request self.request_tracker.record_request(True, response_time) increment_counter("api_requests_successful") record_histogram("api_response_time", response_time) - + # Parse JSON response try: json_data = response.json() @@ -301,24 +232,24 @@ class EnhancedWaterMonitorScraper: self.request_tracker.record_request(False, response_time, "json_parse_error") increment_counter("api_requests_failed") return None - + water_data = [] - + # Parse JSON data - if json_data and isinstance(json_data, dict) and 'rows' in json_data: - for row in json_data['rows']: + if json_data and isinstance(json_data, dict) and "rows" in json_data: + for row in json_data["rows"]: try: # Parse timestamp - time_str = row.get('hourlytime', '') + time_str = row.get("hourlytime", "") if not time_str: continue - + try: # Format: "1.00", "2.00", ..., "24.00" api_hour = int(float(time_str)) if api_hour < 1 or api_hour > 24: continue - + if api_hour == 24: # Hour 24 = midnight (00:00) of the next day data_time = target_date.replace(hour=0, minute=0, second=0, microsecond=0) @@ -326,17 +257,17 @@ class EnhancedWaterMonitorScraper: else: # Hours 1-23 = 01:00-23:00 of the same day data_time = target_date.replace(hour=api_hour, minute=0, second=0, microsecond=0) - + except (ValueError, IndexError): logger.warning(f"Could not parse timestamp: {time_str}") continue - + # Parse all water levels and discharge values station_count = 0 for station_num in range(1, 17): # Stations 1-16 - wl_key = f'wlvalues{station_num}' - q_key = f'qvalues{station_num}' - qp_key = f'QPercent{station_num}' + wl_key = f"wlvalues{station_num}" + q_key = f"qvalues{station_num}" + qp_key = f"QPercent{station_num}" # Check if water level data exists (required) if wl_key in row: @@ -368,51 +299,64 @@ class EnhancedWaterMonitorScraper: except (ValueError, TypeError): discharge_percent = None else: - logger.debug(f"Skipping malformed discharge data for station {station_num}: {discharge_raw}") + logger.debug( + "Skipping malformed discharge data for " + f"station {station_num}: {discharge_raw}" + ) except (ValueError, TypeError) as e: - logger.debug(f"Could not parse discharge for station {station_num}: {e}") + logger.debug( + f"Could not parse discharge for station {station_num}: {e}" + ) - station_info = self.station_mapping.get(str(station_num), { - 'code': f'P.{19+station_num}', - 'thai_name': f'Station {station_num}', - 'english_name': f'Station {station_num}' - }) + station_info = self.station_mapping.get( + str(station_num), + { + "code": f"P.{19+station_num}", + "thai_name": f"Station {station_num}", + "english_name": f"Station {station_num}", + }, + ) - water_data.append({ - 'timestamp': data_time, - 'station_id': station_num, - 'station_code': station_info['code'], - 'station_name_en': station_info['english_name'], - 'station_name_th': station_info['thai_name'], - 'latitude': station_info.get('latitude'), - 'longitude': station_info.get('longitude'), - 'geohash': station_info.get('geohash'), - 'water_level': water_level, - 'water_level_unit': 'm', - 'discharge': discharge, - 'discharge_unit': 'cms', - 'discharge_percent': discharge_percent, - 'status': 'active' - }) + water_data.append( + { + "timestamp": data_time, + "station_id": station_num, + "station_code": station_info["code"], + "station_name_en": station_info["english_name"], + "station_name_th": station_info["thai_name"], + "latitude": station_info.get("latitude"), + "longitude": station_info.get("longitude"), + "geohash": station_info.get("geohash"), + "water_level": water_level, + "water_level_unit": "m", + "discharge": discharge, + "discharge_unit": "cms", + "discharge_percent": discharge_percent, + "status": "active", + } + ) station_count += 1 except (ValueError, TypeError) as e: logger.warning(f"Could not parse water level for station {station_num}: {e}") continue - + logger.debug(f"Processed {station_count} stations for time {time_str}") - + except Exception as e: logger.warning(f"Error processing data row: {e}") continue - + # Validate data water_data = DataValidator.validate_measurements(water_data) - - logger.info(f"Successfully fetched {len(water_data)} data points from API for {target_date.strftime('%Y-%m-%d')}") + + logger.info( + f"Successfully fetched {len(water_data)} data points from API " + f"for {target_date.strftime('%Y-%m-%d')}" + ) return water_data - + except requests.RequestException as e: logger.error(f"Network error fetching API data: {e}") self.request_tracker.record_request(False, 0, "network_error") @@ -453,17 +397,17 @@ class EnhancedWaterMonitorScraper: logger.info("Before 01:00 - fetching yesterday's data only") yesterday = current_time - datetime.timedelta(days=1) return self.fetch_water_data_for_date(yesterday) - + def save_to_database(self, water_data: List[Dict], max_retries: int = 3) -> bool: """Save water measurements to database with retry logic""" if not self.db_adapter: logger.error("Database adapter not initialized") return False - + if not water_data: logger.warning("No data to save") return False - + for attempt in range(max_retries): try: success = self.db_adapter.save_measurements(water_data) @@ -474,31 +418,31 @@ class EnhancedWaterMonitorScraper: return True else: logger.warning(f"Save attempt {attempt + 1} failed, retrying...") - + except Exception as e: if "database is locked" in str(e).lower() and attempt < max_retries - 1: logger.warning(f"Database locked on attempt {attempt + 1}, retrying in {2 ** attempt} seconds...") - time.sleep(2 ** attempt) # Exponential backoff + time.sleep(2**attempt) # Exponential backoff continue else: logger.error(f"Error saving to database (attempt {attempt + 1}): {e}") if attempt == max_retries - 1: increment_counter("database_saves_failed") return False - + return False - + def get_latest_data(self, limit: int = 100) -> List[Dict]: """Get latest data from database""" if not self.db_adapter: return [] - + try: return self.db_adapter.get_latest_measurements(limit=limit) except Exception as e: 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: @@ -510,7 +454,7 @@ class EnhancedWaterMonitorScraper: # Find the most recent timestamp in the data latest_timestamp = None for data_point in water_data: - timestamp = data_point.get('timestamp') + timestamp = data_point.get("timestamp") if timestamp and (latest_timestamp is None or timestamp > latest_timestamp): latest_timestamp = timestamp @@ -522,7 +466,9 @@ class EnhancedWaterMonitorScraper: 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 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 @@ -666,8 +612,12 @@ class EnhancedWaterMonitorScraper: logger.debug(f"Error checking data existence: {e}") return False - def import_historical_data(self, start_date: datetime.datetime, end_date: datetime.datetime, - skip_existing: bool = True) -> int: + def import_historical_data( + self, + start_date: datetime.datetime, + end_date: datetime.datetime, + skip_existing: bool = True, + ) -> int: """ Import historical data for a date range @@ -718,35 +668,28 @@ class EnhancedWaterMonitorScraper: logger.info(f"Historical import completed. Total data points imported: {total_imported}") return total_imported + # Main execution for standalone usage if __name__ == "__main__": import argparse import sys - - # Configure basic logging for standalone usage - import logging + logging.basicConfig( level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=[ - logging.FileHandler('water_monitor.log'), - logging.StreamHandler() - ] + format="%(asctime)s - %(levelname)s - %(message)s", + handlers=[logging.FileHandler("water_monitor.log"), logging.StreamHandler()], ) - + parser = argparse.ArgumentParser(description="Thailand Water Monitor") parser.add_argument("--test", action="store_true", help="Run single test cycle") args = parser.parse_args() - + # Default SQLite configuration - db_config = { - 'type': 'sqlite', - 'connection_string': 'sqlite:///water_levels.db' - } - + db_config = {"type": "sqlite", "connection_string": "sqlite:///water_levels.db"} + try: scraper = EnhancedWaterMonitorScraper(db_config) - + if args.test: logger.info("Running test cycle...") result = scraper.run_scraping_cycle() @@ -759,16 +702,16 @@ if __name__ == "__main__": else: logger.info("Starting continuous monitoring...") schedule.every(1).hours.do(scraper.run_scraping_cycle) - + # Run initial cycle scraper.run_scraping_cycle() - + while True: schedule.run_pending() time.sleep(60) - + except KeyboardInterrupt: logger.info("Monitoring stopped by user") except Exception as e: logger.error(f"Error: {e}") - sys.exit(1) \ No newline at end of file + sys.exit(1) diff --git a/src/web_api.py b/src/web_api.py index 82c1a23..19e2391 100644 --- a/src/web_api.py +++ b/src/web_api.py @@ -353,8 +353,9 @@ async def create_station(station: StationCreateModel): existing_ids = [int(sid) for sid in scraper.station_mapping.keys()] new_station_id = max(existing_ids) + 1 if existing_ids else 1 - # Add to station mapping - scraper.station_mapping[str(new_station_id)] = { + # Add to station mapping and persist + new_key = str(new_station_id) + scraper.station_mapping[new_key] = { "code": station.station_code, "thai_name": station.thai_name, "english_name": station.english_name, @@ -362,6 +363,9 @@ async def create_station(station: StationCreateModel): "longitude": station.longitude, "geohash": station.geohash, } + if not scraper.save_stations(): + scraper.station_mapping.pop(new_key, None) + raise HTTPException(status_code=500, detail="Failed to persist new station") logger.info(f"Created new station: {station.station_code} ({station.english_name})") @@ -376,6 +380,8 @@ async def create_station(station: StationCreateModel): status=station.status, ) + except HTTPException: + raise except Exception as e: logger.error(f"Error creating station: {e}") raise HTTPException(status_code=500, detail=str(e)) @@ -396,6 +402,7 @@ async def update_station(station_id: int, updates: StationUpdateModel): try: station_info = scraper.station_mapping[station_key] + original = dict(station_info) # snapshot for rollback if persistence fails # Update fields if provided if updates.thai_name is not None: @@ -409,6 +416,10 @@ async def update_station(station_id: int, updates: StationUpdateModel): if updates.geohash is not None: station_info["geohash"] = updates.geohash + if not scraper.save_stations(): + scraper.station_mapping[station_key] = original + raise HTTPException(status_code=500, detail="Failed to persist station update") + logger.info(f"Updated station {station_id}: {station_info['code']}") return StationResponse( @@ -422,6 +433,8 @@ async def update_station(station_id: int, updates: StationUpdateModel): status=updates.status or "active", ) + except HTTPException: + raise except Exception as e: logger.error(f"Error updating station {station_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) @@ -442,10 +455,17 @@ async def delete_station(station_id: int): try: station_info = scraper.station_mapping.pop(station_key) + + if not scraper.save_stations(): + scraper.station_mapping[station_key] = station_info # restore + raise HTTPException(status_code=500, detail="Failed to persist station deletion") + logger.info(f"Deleted station {station_id}: {station_info['code']}") return {"message": f"Station {station_info['code']} deleted successfully"} + except HTTPException: + raise except Exception as e: logger.error(f"Error deleting station {station_id}: {e}") raise HTTPException(status_code=500, detail=str(e))