diff --git a/src/alerting.py b/src/alerting.py index 6361bce..76dda11 100644 --- a/src/alerting.py +++ b/src/alerting.py @@ -4,7 +4,9 @@ Water Level Alerting System with Matrix Integration """ import datetime +import html import os +import re from dataclasses import dataclass from enum import Enum from typing import Dict, List, Optional @@ -28,6 +30,31 @@ except ImportError: logger = get_logger(__name__) +_BOLD_RE = re.compile(r"\*\*(.+?)\*\*") +_URL_RE = re.compile(r"(https?://[^\s<]+)") + + +def markdown_to_matrix_html(text: str) -> str: + """Convert the small Markdown subset we emit into Matrix-compatible HTML. + + Matrix clients do NOT render Markdown in the plain ``body`` field; formatting + only shows when an HTML ``formatted_body`` is sent alongside it. We only use + ``**bold**``, bare URLs and newlines, so a minimal converter is sufficient and + avoids adding a Markdown dependency. + """ + # Escape HTML special chars first so station/message data can't inject markup. + result = html.escape(text, quote=False) + result = _BOLD_RE.sub(r"\1", result) + result = _URL_RE.sub(r'\1', result) + result = result.replace("\n", "
") + return result + + +def strip_markdown(text: str) -> str: + """Produce a clean plain-text fallback for the Matrix ``body`` field.""" + return _BOLD_RE.sub(r"\1", text) + + class AlertLevel(Enum): INFO = "info" WARNING = "warning" @@ -55,16 +82,32 @@ class MatrixNotifier: self.room_id = room_id self.session = requests.Session() - def send_message(self, message: str, msgtype: str = "m.text") -> bool: - """Send message to Matrix room""" + def send_message(self, message: str, msgtype: str = "m.text", markdown: bool = True) -> bool: + """Send a message to the Matrix room. + + When ``markdown`` is True (default) the ``message`` is treated as Markdown: + a rendered HTML ``formatted_body`` is sent so clients show real formatting, + with a plain-text ``body`` fallback for clients that ignore HTML. + """ try: # Add transaction ID to prevent duplicates txn_id = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f") url = f"{self.homeserver}/_matrix/client/v3/rooms/{self.room_id}/send/m.room.message/{txn_id}" - headers = {"Authorization": f"Bearer {self.access_token}", "Content-Type": "application/json"} + headers = { + "Authorization": f"Bearer {self.access_token}", + "Content-Type": "application/json", + } - data = {"msgtype": msgtype, "body": message} + if markdown: + data = { + "msgtype": msgtype, + "body": strip_markdown(message), + "format": "org.matrix.custom.html", + "formatted_body": markdown_to_matrix_html(message), + } + else: + data = {"msgtype": msgtype, "body": message} # Matrix API requires PUT when transaction ID is in the URL path response = self.session.put(url, headers=headers, json=data, timeout=10) @@ -239,7 +282,12 @@ class WaterLevelAlertSystem: ("zone_1", 3.7, AlertLevel.INFO, "Zone 1 - Info"), ] - for zone_name, zone_threshold, zone_alert_level, zone_description in zones: + for ( + zone_name, + zone_threshold, + zone_alert_level, + zone_description, + ) in zones: if water_level >= zone_threshold: alert_level = zone_alert_level threshold_value = zone_threshold @@ -348,7 +396,9 @@ class WaterLevelAlertSystem: # Get measurements for this station in the time window current_time = datetime.datetime.now() measurements = self.db_adapter.get_measurements_by_timerange( - start_time=cutoff_time, end_time=current_time, station_codes=[station_code] + start_time=cutoff_time, + end_time=current_time, + station_codes=[station_code], ) if len(measurements) < 2: diff --git a/src/config.py b/src/config.py index 600a583..8b8bd2c 100644 --- a/src/config.py +++ b/src/config.py @@ -1,9 +1,10 @@ import os -from typing import Dict, Any, Optional +from typing import Any, Dict # Load environment variables from .env file try: from dotenv import load_dotenv + load_dotenv() except ImportError: # python-dotenv not installed, continue without it @@ -11,14 +12,14 @@ except ImportError: try: from .exceptions import ConfigurationError - from .models import DatabaseType, DatabaseConfig + from .models import DatabaseType except ImportError: # Handle case when running as standalone script class ConfigurationError(Exception): pass - + from enum import Enum - + class DatabaseType(Enum): SQLITE = "sqlite" MYSQL = "mysql" @@ -26,87 +27,100 @@ except ImportError: INFLUXDB = "influxdb" VICTORIAMETRICS = "victoriametrics" + class Config: """Configuration class for the Water Level Monitor""" - + # Database settings - DATABASE_PATH = os.getenv('WATER_DB_PATH', 'water_levels.db') - + DATABASE_PATH = os.getenv("WATER_DB_PATH", "water_levels.db") + # Website settings TARGET_URL = "https://hyd-app-db.rid.go.th/hydro1h.html" API_URL = "https://hyd-app-db.rid.go.th/webservice/getGroupHourlyWaterLevelReportAllHL.ashx" - REQUEST_TIMEOUT = int(os.getenv('REQUEST_TIMEOUT', '30')) - 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" - + REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "30")) + 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" + ) + # Database configuration - DB_TYPE = os.getenv('DB_TYPE', 'sqlite').lower() - + DB_TYPE = os.getenv("DB_TYPE", "sqlite").lower() + # VictoriaMetrics settings - VM_HOST = os.getenv('VM_HOST', 'vm.newedge.house') - VM_PORT = int(os.getenv('VM_PORT', '443')) - + # Default to localhost; set VM_HOST in the environment for real deployments + # (avoids committing infrastructure hostnames to the repo). + VM_HOST = os.getenv("VM_HOST", "localhost") + VM_PORT = int(os.getenv("VM_PORT", "443")) + # Support for HTTPS URLs (e.g., behind reverse proxy) - VM_URL = os.getenv('VM_URL') # Full URL override (e.g., https://vm.example.com) - + VM_URL = os.getenv("VM_URL") # Full URL override (e.g., https://vm.example.com) + # InfluxDB settings - INFLUX_HOST = os.getenv('INFLUX_HOST', 'localhost') - INFLUX_PORT = int(os.getenv('INFLUX_PORT', '8086')) - INFLUX_DATABASE = os.getenv('INFLUX_DATABASE', 'water_monitoring') - INFLUX_USERNAME = os.getenv('INFLUX_USERNAME') - INFLUX_PASSWORD = os.getenv('INFLUX_PASSWORD') - + INFLUX_HOST = os.getenv("INFLUX_HOST", "localhost") + INFLUX_PORT = int(os.getenv("INFLUX_PORT", "8086")) + INFLUX_DATABASE = os.getenv("INFLUX_DATABASE", "water_monitoring") + INFLUX_USERNAME = os.getenv("INFLUX_USERNAME") + INFLUX_PASSWORD = os.getenv("INFLUX_PASSWORD") + # PostgreSQL settings - POSTGRES_CONNECTION_STRING = os.getenv('POSTGRES_CONNECTION_STRING') - POSTGRES_HOST = os.getenv('POSTGRES_HOST', 'localhost') - POSTGRES_PORT = int(os.getenv('POSTGRES_PORT', '5432')) - POSTGRES_DB = os.getenv('POSTGRES_DB', 'water_monitoring') - POSTGRES_USER = os.getenv('POSTGRES_USER', 'postgres') - POSTGRES_PASSWORD = os.getenv('POSTGRES_PASSWORD') - + POSTGRES_CONNECTION_STRING = os.getenv("POSTGRES_CONNECTION_STRING") + POSTGRES_HOST = os.getenv("POSTGRES_HOST", "localhost") + POSTGRES_PORT = int(os.getenv("POSTGRES_PORT", "5432")) + POSTGRES_DB = os.getenv("POSTGRES_DB", "water_monitoring") + POSTGRES_USER = os.getenv("POSTGRES_USER", "postgres") + POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD") + # MySQL settings - MYSQL_CONNECTION_STRING = os.getenv('MYSQL_CONNECTION_STRING') - + MYSQL_CONNECTION_STRING = os.getenv("MYSQL_CONNECTION_STRING") + # Scheduler settings - SCRAPING_INTERVAL_HOURS = int(os.getenv('SCRAPING_INTERVAL_HOURS', '1')) - + SCRAPING_INTERVAL_HOURS = int(os.getenv("SCRAPING_INTERVAL_HOURS", "1")) + # Logging settings - LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO') - LOG_FILE = os.getenv('LOG_FILE', 'water_monitor.log') - LOG_FORMAT = '%(asctime)s - %(levelname)s - %(message)s' - + LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO") + LOG_FILE = os.getenv("LOG_FILE", "water_monitor.log") + LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s" + # Data retention - DATA_RETENTION_DAYS = int(os.getenv('DATA_RETENTION_DAYS', '365')) - + DATA_RETENTION_DAYS = int(os.getenv("DATA_RETENTION_DAYS", "365")) + # Retry settings - MAX_RETRIES = int(os.getenv('MAX_RETRIES', '3')) - RETRY_DELAY_SECONDS = int(os.getenv('RETRY_DELAY_SECONDS', '60')) - + MAX_RETRIES = int(os.getenv("MAX_RETRIES", "3")) + RETRY_DELAY_SECONDS = int(os.getenv("RETRY_DELAY_SECONDS", "60")) + + # 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. + # Credentials are only enabled when explicit (non-wildcard) origins are set, + # because "*" + credentials is rejected by browsers and unsafe. + CORS_ALLOW_ORIGINS = [origin.strip() for origin in os.getenv("CORS_ALLOW_ORIGINS", "").split(",") if origin.strip()] + @classmethod def validate_config(cls) -> bool: """Validate configuration settings""" errors = [] - + # Validate database type try: DatabaseType(cls.DB_TYPE) except ValueError: errors.append(f"Invalid DB_TYPE: {cls.DB_TYPE}") - + # Validate database-specific settings - if cls.DB_TYPE == 'victoriametrics': + if cls.DB_TYPE == "victoriametrics": if not cls.VM_HOST: errors.append("VM_HOST is required for VictoriaMetrics") if not isinstance(cls.VM_PORT, int) or cls.VM_PORT <= 0: errors.append("VM_PORT must be a positive integer") - - elif cls.DB_TYPE == 'influxdb': + + elif cls.DB_TYPE == "influxdb": if not cls.INFLUX_HOST: errors.append("INFLUX_HOST is required for InfluxDB") if not cls.INFLUX_DATABASE: errors.append("INFLUX_DATABASE is required for InfluxDB") - - elif cls.DB_TYPE in ['postgresql', 'mysql']: - if cls.DB_TYPE == 'postgresql': + + elif cls.DB_TYPE in ["postgresql", "mysql"]: + if cls.DB_TYPE == "postgresql": # Check if either connection string or individual components are provided if not cls.POSTGRES_CONNECTION_STRING: # If no connection string, check individual components @@ -121,105 +135,103 @@ class Config: else: # mysql if not cls.MYSQL_CONNECTION_STRING: errors.append("MYSQL_CONNECTION_STRING is required for MySQL") - + # Validate numeric settings if cls.SCRAPING_INTERVAL_HOURS <= 0: errors.append("SCRAPING_INTERVAL_HOURS must be positive") - + if cls.DATA_RETENTION_DAYS <= 0: errors.append("DATA_RETENTION_DAYS must be positive") - + if errors: raise ConfigurationError(f"Configuration errors: {'; '.join(errors)}") - + return True - + @classmethod def get_database_config(cls) -> Dict[str, Any]: """Returns database configuration based on DB_TYPE""" - if cls.DB_TYPE == 'victoriametrics': + if cls.DB_TYPE == "victoriametrics": + return {"type": "victoriametrics", "host": cls.VM_HOST, "port": cls.VM_PORT} + elif cls.DB_TYPE == "influxdb": return { - 'type': 'victoriametrics', - 'host': cls.VM_HOST, - 'port': cls.VM_PORT + "type": "influxdb", + "host": cls.INFLUX_HOST, + "port": cls.INFLUX_PORT, + "database": cls.INFLUX_DATABASE, + "username": cls.INFLUX_USERNAME, + "password": cls.INFLUX_PASSWORD, } - elif cls.DB_TYPE == 'influxdb': - return { - 'type': 'influxdb', - 'host': cls.INFLUX_HOST, - 'port': cls.INFLUX_PORT, - 'database': cls.INFLUX_DATABASE, - 'username': cls.INFLUX_USERNAME, - 'password': cls.INFLUX_PASSWORD - } - elif cls.DB_TYPE == 'postgresql': + elif cls.DB_TYPE == "postgresql": # Use individual components if POSTGRES_CONNECTION_STRING is not provided if cls.POSTGRES_CONNECTION_STRING: return { - 'type': 'postgresql', - 'connection_string': cls.POSTGRES_CONNECTION_STRING + "type": "postgresql", + "connection_string": cls.POSTGRES_CONNECTION_STRING, } else: # Build connection string from components (automatically URL-encodes password) import urllib.parse - password = urllib.parse.quote(cls.POSTGRES_PASSWORD or 'password', safe='') - connection_string = f'postgresql://{cls.POSTGRES_USER}:{password}@{cls.POSTGRES_HOST}:{cls.POSTGRES_PORT}/{cls.POSTGRES_DB}' - return { - 'type': 'postgresql', - 'connection_string': connection_string - } - elif cls.DB_TYPE == 'mysql': - return { - 'type': 'mysql', - 'connection_string': cls.MYSQL_CONNECTION_STRING or - 'mysql://root:password@localhost:3306/water_monitoring' - } + + if not cls.POSTGRES_PASSWORD: + raise ConfigurationError("POSTGRES_PASSWORD is required for PostgreSQL (no default is provided)") + password = urllib.parse.quote(cls.POSTGRES_PASSWORD, safe="") + connection_string = ( + f"postgresql://{cls.POSTGRES_USER}:{password}" + f"@{cls.POSTGRES_HOST}:{cls.POSTGRES_PORT}/{cls.POSTGRES_DB}" + ) + return {"type": "postgresql", "connection_string": connection_string} + elif cls.DB_TYPE == "mysql": + if not cls.MYSQL_CONNECTION_STRING: + raise ConfigurationError("MYSQL_CONNECTION_STRING is required for MySQL (no default is provided)") + return {"type": "mysql", "connection_string": cls.MYSQL_CONNECTION_STRING} else: # sqlite return { - 'type': 'sqlite', - 'connection_string': f'sqlite:///{cls.DATABASE_PATH}' + "type": "sqlite", + "connection_string": f"sqlite:///{cls.DATABASE_PATH}", } - + @classmethod def get_all_settings(cls) -> Dict[str, Any]: """Returns all configuration settings""" return { - 'DB_TYPE': cls.DB_TYPE, - 'DATABASE_PATH': cls.DATABASE_PATH, - 'TARGET_URL': cls.TARGET_URL, - 'API_URL': cls.API_URL, - 'REQUEST_TIMEOUT': cls.REQUEST_TIMEOUT, - 'SCRAPING_INTERVAL_HOURS': cls.SCRAPING_INTERVAL_HOURS, - 'LOG_LEVEL': cls.LOG_LEVEL, - 'LOG_FILE': cls.LOG_FILE, - 'DATA_RETENTION_DAYS': cls.DATA_RETENTION_DAYS, - 'MAX_RETRIES': cls.MAX_RETRIES, - 'RETRY_DELAY_SECONDS': cls.RETRY_DELAY_SECONDS, - 'VM_HOST': cls.VM_HOST, - 'VM_PORT': cls.VM_PORT, - 'INFLUX_HOST': cls.INFLUX_HOST, - 'INFLUX_PORT': cls.INFLUX_PORT, - 'INFLUX_DATABASE': cls.INFLUX_DATABASE + "DB_TYPE": cls.DB_TYPE, + "DATABASE_PATH": cls.DATABASE_PATH, + "TARGET_URL": cls.TARGET_URL, + "API_URL": cls.API_URL, + "REQUEST_TIMEOUT": cls.REQUEST_TIMEOUT, + "SCRAPING_INTERVAL_HOURS": cls.SCRAPING_INTERVAL_HOURS, + "LOG_LEVEL": cls.LOG_LEVEL, + "LOG_FILE": cls.LOG_FILE, + "DATA_RETENTION_DAYS": cls.DATA_RETENTION_DAYS, + "MAX_RETRIES": cls.MAX_RETRIES, + "RETRY_DELAY_SECONDS": cls.RETRY_DELAY_SECONDS, + "VM_HOST": cls.VM_HOST, + "VM_PORT": cls.VM_PORT, + "INFLUX_HOST": cls.INFLUX_HOST, + "INFLUX_PORT": cls.INFLUX_PORT, + "INFLUX_DATABASE": cls.INFLUX_DATABASE, } - + @classmethod def print_settings(cls): """Prints all current settings""" print("=== Water Level Monitor Configuration ===") for key, value in cls.get_all_settings().items(): # Hide sensitive information - if 'PASSWORD' in key and value: - value = '*' * len(str(value)) + if "PASSWORD" in key and value: + value = "*" * len(str(value)) print(f"{key}: {value}") print("=" * 45) - + print("\nDatabase Configuration:") db_config = cls.get_database_config() for key, value in db_config.items(): - if 'password' in key and value: - value = '*' * len(str(value)) + if "password" in key and value: + value = "*" * len(str(value)) print(f" {key}: {value}") print("=" * 45) + if __name__ == "__main__": Config.print_settings() diff --git a/src/database_adapters.py b/src/database_adapters.py index 944154a..588a2c5 100644 --- a/src/database_adapters.py +++ b/src/database_adapters.py @@ -5,225 +5,255 @@ Database adapters for different storage backends import datetime import logging -from typing import List, Dict, Optional, Any from abc import ABC, abstractmethod +from typing import Dict, List, Optional + # Base adapter interface class DatabaseAdapter(ABC): @abstractmethod def connect(self): pass - + @abstractmethod def save_measurements(self, measurements: List[Dict]) -> bool: pass - + @abstractmethod def get_latest_measurements(self, limit: int = 100) -> List[Dict]: pass - + @abstractmethod - def get_measurements_by_timerange(self, start_time: datetime.datetime, - end_time: datetime.datetime, - station_codes: Optional[List[str]] = None) -> List[Dict]: + def get_measurements_by_timerange( + self, + start_time: datetime.datetime, + end_time: datetime.datetime, + station_codes: Optional[List[str]] = None, + ) -> List[Dict]: pass @abstractmethod def get_measurements_for_date(self, target_date: datetime.datetime) -> List[Dict]: pass + # InfluxDB Adapter class InfluxDBAdapter(DatabaseAdapter): - def __init__(self, host: str = "localhost", port: int = 8086, - database: str = "water_monitoring", username: str = None, password: str = None): + def __init__( + self, + host: str = "localhost", + port: int = 8086, + database: str = "water_monitoring", + username: str = None, + password: str = None, + ): self.host = host self.port = port self.database = database self.username = username self.password = password self.client = None - + def connect(self): try: from influxdb import InfluxDBClient + self.client = InfluxDBClient( host=self.host, port=self.port, username=self.username, password=self.password, - database=self.database + database=self.database, ) - + # Create database if it doesn't exist databases = self.client.get_list_database() - if not any(db['name'] == self.database for db in databases): + if not any(db["name"] == self.database for db in databases): self.client.create_database(self.database) logging.info(f"Created InfluxDB database: {self.database}") - + # Create retention policy (keep data for 2 years, downsample after 30 days) retention_policies = self.client.get_list_retention_policies(self.database) - if not any(rp['name'] == 'water_data_policy' for rp in retention_policies): + if not any(rp["name"] == "water_data_policy" for rp in retention_policies): self.client.create_retention_policy( - 'water_data_policy', - '730d', # 2 years - '1', # replication factor + "water_data_policy", + "730d", # 2 years + "1", # replication factor database=self.database, - default=True + default=True, ) - + logging.info("Connected to InfluxDB successfully") return True - + except ImportError: logging.error("InfluxDB client not installed. Run: pip install influxdb") return False except Exception as e: logging.error(f"Failed to connect to InfluxDB: {e}") return False - + def save_measurements(self, measurements: List[Dict]) -> bool: if not self.client: logging.error("InfluxDB client not connected") return False - + try: points = [] for measurement in measurements: point = { "measurement": "water_data", "tags": { - "station_code": measurement['station_code'], - "station_name_en": measurement['station_name_en'], - "station_name_th": measurement['station_name_th'] + "station_code": measurement["station_code"], + "station_name_en": measurement["station_name_en"], + "station_name_th": measurement["station_name_th"], }, - "time": measurement['timestamp'].isoformat(), + "time": measurement["timestamp"].isoformat(), "fields": { - "water_level": float(measurement['water_level']), - "discharge": float(measurement['discharge']), - "discharge_percent": float(measurement['discharge_percent']) if measurement['discharge_percent'] else None - } + "water_level": float(measurement["water_level"]), + "discharge": float(measurement["discharge"]), + "discharge_percent": float(measurement["discharge_percent"]) + if measurement["discharge_percent"] + else None, + }, } points.append(point) - + success = self.client.write_points(points) if success: logging.info(f"Successfully wrote {len(points)} points to InfluxDB") return success - + except Exception as e: logging.error(f"Error writing to InfluxDB: {e}") return False - + def get_latest_measurements(self, limit: int = 100) -> List[Dict]: if not self.client: return [] - + try: + # Cast limit to int so it can never carry an injection payload. + limit = int(limit) query = f""" - SELECT last("water_level") as water_level, - last("discharge") as discharge, + SELECT last("water_level") as water_level, + last("discharge") as discharge, last("discharge_percent") as discharge_percent - FROM "water_data" + FROM "water_data" GROUP BY "station_code", "station_name_en", "station_name_th" LIMIT {limit} """ - + result = self.client.query(query) measurements = [] - + for point in result.get_points(): - measurements.append({ - 'timestamp': point['time'], - 'station_code': point.get('station_code'), - 'station_name_en': point.get('station_name_en'), - 'station_name_th': point.get('station_name_th'), - 'water_level': point.get('water_level'), - 'discharge': point.get('discharge'), - 'discharge_percent': point.get('discharge_percent') - }) - + measurements.append( + { + "timestamp": point["time"], + "station_code": point.get("station_code"), + "station_name_en": point.get("station_name_en"), + "station_name_th": point.get("station_name_th"), + "water_level": point.get("water_level"), + "discharge": point.get("discharge"), + "discharge_percent": point.get("discharge_percent"), + } + ) + return measurements - + except Exception as e: logging.error(f"Error querying InfluxDB: {e}") return [] - - def get_measurements_by_timerange(self, start_time: datetime.datetime, - end_time: datetime.datetime, - station_codes: Optional[List[str]] = None) -> List[Dict]: + + def get_measurements_by_timerange( + self, + start_time: datetime.datetime, + end_time: datetime.datetime, + station_codes: Optional[List[str]] = None, + ) -> List[Dict]: if not self.client: return [] - + try: + # start_time/end_time are datetime objects (fixed isoformat, injection-safe). + # station_codes are untrusted strings -> bind them as parameters. + bind_params = {} where_clause = f"time >= '{start_time.isoformat()}' AND time <= '{end_time.isoformat()}'" if station_codes: - station_filter = "'" + "','".join(station_codes) + "'" - where_clause += f" AND station_code IN ({station_filter})" - + placeholders = [] + for i, code in enumerate(station_codes): + key = f"sc{i}" + bind_params[key] = code + placeholders.append(f"station_code = ${key}") + where_clause += " AND (" + " OR ".join(placeholders) + ")" + query = f""" SELECT "water_level", "discharge", "discharge_percent", "station_code", "station_name_en", "station_name_th" - FROM "water_data" + FROM "water_data" WHERE {where_clause} ORDER BY time DESC """ - - result = self.client.query(query) + + result = self.client.query(query, bind_params=bind_params) measurements = [] - + for point in result.get_points(): - measurements.append({ - 'timestamp': point['time'], - 'station_code': point.get('station_code'), - 'station_name_en': point.get('station_name_en'), - 'station_name_th': point.get('station_name_th'), - 'water_level': point.get('water_level'), - 'discharge': point.get('discharge'), - 'discharge_percent': point.get('discharge_percent') - }) - + measurements.append( + { + "timestamp": point["time"], + "station_code": point.get("station_code"), + "station_name_en": point.get("station_name_en"), + "station_name_th": point.get("station_name_th"), + "water_level": point.get("water_level"), + "discharge": point.get("discharge"), + "discharge_percent": point.get("discharge_percent"), + } + ) + return measurements - + except Exception as e: logging.error(f"Error querying InfluxDB: {e}") return [] + # MySQL/PostgreSQL Adapter class SQLAdapter(DatabaseAdapter): def __init__(self, connection_string: str, db_type: str = "mysql"): self.connection_string = connection_string self.db_type = db_type.lower() self.engine = None - + # Add SQLite-specific connection parameters for better concurrency if self.db_type == "sqlite": if "?" not in connection_string: self.connection_string += "?timeout=30&check_same_thread=False" else: self.connection_string += "&timeout=30&check_same_thread=False" - + def connect(self): try: - from sqlalchemy import create_engine, text - from sqlalchemy.orm import sessionmaker - + from sqlalchemy import create_engine + self.engine = create_engine(self.connection_string, pool_pre_ping=True) - + # Create tables self._create_tables() - + logging.info(f"Connected to {self.db_type.upper()} successfully") return True - + except ImportError: logging.error("SQLAlchemy not installed. Run: pip install sqlalchemy pymysql") return False except Exception as e: logging.error(f"Failed to connect to {self.db_type.upper()}: {e}") return False - + def _create_tables(self): from sqlalchemy import text - + # Stations table - adjust for different databases if self.db_type == "sqlite": stations_sql = """ @@ -239,7 +269,7 @@ class SQLAdapter(DatabaseAdapter): updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """ - + measurements_sql = """ CREATE TABLE IF NOT EXISTS water_measurements ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -254,13 +284,13 @@ class SQLAdapter(DatabaseAdapter): UNIQUE(timestamp, station_id) ) """ - + # Create indexes separately for SQLite index_sql = [ "CREATE INDEX IF NOT EXISTS idx_timestamp ON water_measurements(timestamp)", - "CREATE INDEX IF NOT EXISTS idx_station_timestamp ON water_measurements(station_id, timestamp)" + "CREATE INDEX IF NOT EXISTS idx_station_timestamp ON water_measurements(station_id, timestamp)", ] - + elif self.db_type == "postgresql": stations_sql = """ CREATE TABLE IF NOT EXISTS stations ( @@ -275,7 +305,7 @@ class SQLAdapter(DatabaseAdapter): updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """ - + measurements_sql = """ CREATE TABLE IF NOT EXISTS water_measurements ( id BIGSERIAL PRIMARY KEY, @@ -290,12 +320,12 @@ class SQLAdapter(DatabaseAdapter): UNIQUE(timestamp, station_id) ) """ - + index_sql = [ "CREATE INDEX IF NOT EXISTS idx_timestamp ON water_measurements(timestamp)", - "CREATE INDEX IF NOT EXISTS idx_station_timestamp ON water_measurements(station_id, timestamp DESC)" + "CREATE INDEX IF NOT EXISTS idx_station_timestamp ON water_measurements(station_id, timestamp DESC)", ] - + else: # MySQL stations_sql = """ CREATE TABLE IF NOT EXISTS stations ( @@ -310,7 +340,7 @@ class SQLAdapter(DatabaseAdapter): updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) """ - + measurements_sql = """ CREATE TABLE IF NOT EXISTS water_measurements ( id BIGINT AUTO_INCREMENT PRIMARY KEY, @@ -328,36 +358,44 @@ class SQLAdapter(DatabaseAdapter): ) """ index_sql = [] - + with self.engine.begin() as conn: conn.execute(text(stations_sql)) conn.execute(text(measurements_sql)) - + # Create indexes for SQLite and PostgreSQL for index in index_sql: conn.execute(text(index)) - + # Transaction is automatically committed when context manager exits - + def save_measurements(self, measurements: List[Dict]) -> bool: if not self.engine: return False - + try: from sqlalchemy import text - + with self.engine.begin() as conn: # Insert/update stations for measurement in measurements: if self.db_type == "sqlite": station_sql = """ - INSERT OR REPLACE INTO stations (id, station_code, thai_name, english_name, latitude, longitude, geohash, updated_at) - VALUES (:station_id, :station_code, :thai_name, :english_name, :latitude, :longitude, :geohash, CURRENT_TIMESTAMP) + INSERT OR REPLACE INTO stations + (id, station_code, thai_name, english_name, + latitude, longitude, geohash, updated_at) + VALUES + (:station_id, :station_code, :thai_name, :english_name, + :latitude, :longitude, :geohash, CURRENT_TIMESTAMP) """ elif self.db_type == "postgresql": station_sql = """ - INSERT INTO stations (id, station_code, thai_name, english_name, latitude, longitude, geohash, updated_at) - VALUES (:station_id, :station_code, :thai_name, :english_name, :latitude, :longitude, :geohash, NOW()) + INSERT INTO stations + (id, station_code, thai_name, english_name, + latitude, longitude, geohash, updated_at) + VALUES + (:station_id, :station_code, :thai_name, :english_name, + :latitude, :longitude, :geohash, NOW()) ON CONFLICT (id) DO UPDATE SET thai_name = EXCLUDED.thai_name, english_name = EXCLUDED.english_name, @@ -368,9 +406,13 @@ class SQLAdapter(DatabaseAdapter): """ else: # MySQL station_sql = """ - INSERT INTO stations (id, station_code, thai_name, english_name, latitude, longitude, geohash, updated_at) - VALUES (:station_id, :station_code, :thai_name, :english_name, :latitude, :longitude, :geohash, NOW()) - ON DUPLICATE KEY UPDATE + INSERT INTO stations + (id, station_code, thai_name, english_name, + latitude, longitude, geohash, updated_at) + VALUES + (:station_id, :station_code, :thai_name, :english_name, + :latitude, :longitude, :geohash, NOW()) + ON DUPLICATE KEY UPDATE thai_name = VALUES(thai_name), english_name = VALUES(english_name), latitude = VALUES(latitude), @@ -378,28 +420,31 @@ class SQLAdapter(DatabaseAdapter): geohash = VALUES(geohash), updated_at = NOW() """ - - conn.execute(text(station_sql), { - 'station_id': measurement['station_id'], - 'station_code': measurement['station_code'], - 'thai_name': measurement['station_name_th'], - 'english_name': measurement['station_name_en'], - 'latitude': measurement.get('latitude'), - 'longitude': measurement.get('longitude'), - 'geohash': measurement.get('geohash') - }) - + + conn.execute( + text(station_sql), + { + "station_id": measurement["station_id"], + "station_code": measurement["station_code"], + "thai_name": measurement["station_name_th"], + "english_name": measurement["station_name_en"], + "latitude": measurement.get("latitude"), + "longitude": measurement.get("longitude"), + "geohash": measurement.get("geohash"), + }, + ) + # Insert measurements for measurement in measurements: if self.db_type == "sqlite": measurement_sql = """ - INSERT OR REPLACE INTO water_measurements + INSERT OR REPLACE INTO water_measurements (timestamp, station_id, water_level, discharge, discharge_percent, status) VALUES (:timestamp, :station_id, :water_level, :discharge, :discharge_percent, :status) """ elif self.db_type == "postgresql": measurement_sql = """ - INSERT INTO water_measurements + INSERT INTO water_measurements (timestamp, station_id, water_level, discharge, discharge_percent, status) VALUES (:timestamp, :station_id, :water_level, :discharge, :discharge_percent, :status) ON CONFLICT (timestamp, station_id) DO UPDATE SET @@ -410,7 +455,7 @@ class SQLAdapter(DatabaseAdapter): """ else: # MySQL measurement_sql = """ - INSERT INTO water_measurements + INSERT INTO water_measurements (timestamp, station_id, water_level, discharge, discharge_percent, status) VALUES (:timestamp, :station_id, :water_level, :discharge, :discharge_percent, :status) ON DUPLICATE KEY UPDATE @@ -419,31 +464,34 @@ class SQLAdapter(DatabaseAdapter): discharge_percent = VALUES(discharge_percent), status = VALUES(status) """ - - conn.execute(text(measurement_sql), { - 'timestamp': measurement['timestamp'], - 'station_id': measurement['station_id'], - 'water_level': measurement['water_level'], - 'discharge': measurement['discharge'], - 'discharge_percent': measurement['discharge_percent'], - 'status': measurement['status'] - }) - + + conn.execute( + text(measurement_sql), + { + "timestamp": measurement["timestamp"], + "station_id": measurement["station_id"], + "water_level": measurement["water_level"], + "discharge": measurement["discharge"], + "discharge_percent": measurement["discharge_percent"], + "status": measurement["status"], + }, + ) + # Transaction is automatically committed when context manager exits logging.info(f"Successfully saved {len(measurements)} measurements to {self.db_type.upper()}") return True - + except Exception as e: logging.error(f"Error saving to {self.db_type.upper()}: {e}") return False - + def get_latest_measurements(self, limit: int = 100) -> List[Dict]: if not self.engine: return [] - + try: from sqlalchemy import text - + query = """ SELECT m.timestamp, s.station_code, s.english_name, s.thai_name, m.water_level, m.discharge, m.discharge_percent, m.status @@ -457,47 +505,52 @@ class SQLAdapter(DatabaseAdapter): ORDER BY s.station_code LIMIT :limit """ - + with self.engine.connect() as conn: - result = conn.execute(text(query), {'limit': limit}) + result = conn.execute(text(query), {"limit": limit}) measurements = [] - + for row in result: - measurements.append({ - 'timestamp': row[0], - 'station_code': row[1], - 'station_name_en': row[2], - 'station_name_th': row[3], - 'water_level': float(row[4]) if row[4] else None, - 'discharge': float(row[5]) if row[5] else None, - 'discharge_percent': float(row[6]) if row[6] else None, - 'status': row[7] - }) - + measurements.append( + { + "timestamp": row[0], + "station_code": row[1], + "station_name_en": row[2], + "station_name_th": row[3], + "water_level": float(row[4]) if row[4] else None, + "discharge": float(row[5]) if row[5] else None, + "discharge_percent": float(row[6]) if row[6] else None, + "status": row[7], + } + ) + return measurements - + except Exception as e: logging.error(f"Error querying {self.db_type.upper()}: {e}") return [] - - def get_measurements_by_timerange(self, start_time: datetime.datetime, - end_time: datetime.datetime, - station_codes: Optional[List[str]] = None) -> List[Dict]: + + def get_measurements_by_timerange( + self, + start_time: datetime.datetime, + end_time: datetime.datetime, + station_codes: Optional[List[str]] = None, + ) -> List[Dict]: if not self.engine: return [] - + try: from sqlalchemy import text - + where_clause = "m.timestamp BETWEEN :start_time AND :end_time" - params = {'start_time': start_time, 'end_time': end_time} - + params = {"start_time": start_time, "end_time": end_time} + if station_codes: - placeholders = ','.join([f':station_{i}' for i in range(len(station_codes))]) + placeholders = ",".join([f":station_{i}" for i in range(len(station_codes))]) where_clause += f" AND s.station_code IN ({placeholders})" for i, code in enumerate(station_codes): - params[f'station_{i}'] = code - + params[f"station_{i}"] = code + query = f""" SELECT m.timestamp, s.station_code, s.english_name, s.thai_name, m.water_level, m.discharge, m.discharge_percent, m.status @@ -506,25 +559,27 @@ class SQLAdapter(DatabaseAdapter): WHERE {where_clause} ORDER BY m.timestamp DESC, s.station_code """ - + with self.engine.connect() as conn: result = conn.execute(text(query), params) measurements = [] - + for row in result: - measurements.append({ - 'timestamp': row[0], - 'station_code': row[1], - 'station_name_en': row[2], - 'station_name_th': row[3], - 'water_level': float(row[4]) if row[4] else None, - 'discharge': float(row[5]) if row[5] else None, - 'discharge_percent': float(row[6]) if row[6] else None, - 'status': row[7] - }) - + measurements.append( + { + "timestamp": row[0], + "station_code": row[1], + "station_name_en": row[2], + "station_name_th": row[3], + "water_level": float(row[4]) if row[4] else None, + "discharge": float(row[5]) if row[5] else None, + "discharge_percent": float(row[6]) if row[6] else None, + "status": row[7], + } + ) + return measurements - + except Exception as e: logging.error(f"Error querying {self.db_type.upper()}: {e}") return [] @@ -551,23 +606,22 @@ class SQLAdapter(DatabaseAdapter): """ with self.engine.connect() as conn: - result = conn.execute(text(query), { - 'start_time': start_of_day, - 'end_time': end_of_day - }) + result = conn.execute(text(query), {"start_time": start_of_day, "end_time": end_of_day}) measurements = [] for row in result: - measurements.append({ - 'timestamp': row[0], - 'station_id': row[1], - 'station_code': row[2] or f"Station_{row[1]}", - 'station_name_th': row[3] or f"Station {row[1]}", - 'water_level': float(row[4]) if row[4] else None, - 'discharge': float(row[5]) if row[5] else None, - 'discharge_percent': float(row[6]) if row[6] else None, - 'status': row[7] - }) + measurements.append( + { + "timestamp": row[0], + "station_id": row[1], + "station_code": row[2] or f"Station_{row[1]}", + "station_name_th": row[3] or f"Station {row[1]}", + "water_level": float(row[4]) if row[4] else None, + "discharge": float(row[5]) if row[5] else None, + "discharge_percent": float(row[6]) if row[6] else None, + "status": row[7], + } + ) return measurements @@ -575,18 +629,19 @@ class SQLAdapter(DatabaseAdapter): logging.error(f"Error querying {self.db_type.upper()} for date {target_date.date()}: {e}") return [] + # VictoriaMetrics Adapter (using Prometheus format) class VictoriaMetricsAdapter(DatabaseAdapter): def __init__(self, host: str = "localhost", port: int = 8428): self.host = host self.port = port - + # Handle HTTPS URLs and reverse proxy configurations - if host.startswith(('http://', 'https://')): + if host.startswith(("http://", "https://")): self.base_url = host - if port != 80 and port != 443 and not host.endswith(f':{port}'): + if port != 80 and port != 443 and not host.endswith(f":{port}"): # Only add port if it's not standard and not already in URL - if '://' in host and ':' not in host.split('://')[1]: + if "://" in host and ":" not in host.split("://")[1]: self.base_url = f"{host}:{port}" else: # Default to HTTP for localhost, HTTPS for remote hosts @@ -595,15 +650,34 @@ class VictoriaMetricsAdapter(DatabaseAdapter): self.base_url = f"{protocol}://{host}" else: self.base_url = f"{protocol}://{host}:{port}" - + + @staticmethod + def _escape_label(value) -> str: + """Escape a Prometheus label value per the exposition format spec. + + Station names include arbitrary Thai text (and could be set via the API), + so backslashes, double-quotes and newlines must be escaped to avoid + producing malformed or injected exposition lines. + """ + return str(value).replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n") + + @staticmethod + def _metric_value(value) -> Optional[float]: + """Coerce a numeric field to float, or None if it isn't a valid number.""" + try: + return float(value) + except (TypeError, ValueError): + return None + def connect(self): try: import requests + # Test connection with SSL verification and timeout response = requests.get( - f"{self.base_url}/api/v1/status/config", + f"{self.base_url}/api/v1/status/config", timeout=10, - verify=True # Enable SSL verification for HTTPS + verify=True, # Enable SSL verification for HTTPS ) if response.status_code == 200: logging.info(f"Connected to VictoriaMetrics successfully at {self.base_url}") @@ -620,70 +694,70 @@ class VictoriaMetricsAdapter(DatabaseAdapter): except Exception as e: logging.error(f"Failed to connect to VictoriaMetrics: {e}") return False - + def save_measurements(self, measurements: List[Dict]) -> bool: try: import requests - + # Convert to Prometheus format metrics_data = [] timestamp_ms = int(datetime.datetime.now().timestamp() * 1000) - + for measurement in measurements: + # Escape label values once per measurement (untrusted Thai/English names). + labels = ( + f'station_code="{self._escape_label(measurement["station_code"])}",' + f'station_name_en="{self._escape_label(measurement["station_name_en"])}",' + f'station_name_th="{self._escape_label(measurement["station_name_th"])}"' + ) + # Water level metric - metrics_data.append( - f'water_level{{station_code="{measurement["station_code"]}",' - f'station_name_en="{measurement["station_name_en"]}",' - f'station_name_th="{measurement["station_name_th"]}"}} ' - f'{measurement["water_level"]} {timestamp_ms}' - ) - + water_level = self._metric_value(measurement.get("water_level")) + if water_level is not None: + metrics_data.append(f"water_level{{{labels}}} {water_level} {timestamp_ms}") + # Discharge metric - metrics_data.append( - f'water_discharge{{station_code="{measurement["station_code"]}",' - f'station_name_en="{measurement["station_name_en"]}",' - f'station_name_th="{measurement["station_name_th"]}"}} ' - f'{measurement["discharge"]} {timestamp_ms}' - ) - + discharge = self._metric_value(measurement.get("discharge")) + if discharge is not None: + metrics_data.append(f"water_discharge{{{labels}}} {discharge} {timestamp_ms}") + # Discharge percentage metric - if measurement["discharge_percent"]: - metrics_data.append( - f'water_discharge_percent{{station_code="{measurement["station_code"]}",' - f'station_name_en="{measurement["station_name_en"]}",' - f'station_name_th="{measurement["station_name_th"]}"}} ' - f'{measurement["discharge_percent"]} {timestamp_ms}' - ) - + discharge_percent = self._metric_value(measurement.get("discharge_percent")) + if discharge_percent is not None: + metrics_data.append(f"water_discharge_percent{{{labels}}} {discharge_percent} {timestamp_ms}") + # Send to VictoriaMetrics - data = '\n'.join(metrics_data) + data = "\n".join(metrics_data) response = requests.post( f"{self.base_url}/api/v1/import/prometheus", data=data, - headers={'Content-Type': 'text/plain'}, - timeout=30 + headers={"Content-Type": "text/plain"}, + timeout=30, ) - + if response.status_code == 204: logging.info(f"Successfully sent {len(measurements)} measurements to VictoriaMetrics") return True else: logging.error(f"VictoriaMetrics import failed: {response.status_code} - {response.text}") return False - + except Exception as e: logging.error(f"Error sending to VictoriaMetrics: {e}") return False - + def get_latest_measurements(self, limit: int = 100) -> List[Dict]: # VictoriaMetrics queries would be implemented here # This is a simplified version logging.warning("get_latest_measurements not fully implemented for VictoriaMetrics") return [] - - def get_measurements_by_timerange(self, start_time: datetime.datetime, - end_time: datetime.datetime, - station_codes: Optional[List[str]] = None) -> List[Dict]: + + def get_measurements_by_timerange( + self, + start_time: datetime.datetime, + end_time: datetime.datetime, + station_codes: Optional[List[str]] = None, + ) -> List[Dict]: # VictoriaMetrics range queries would be implemented here logging.warning("get_measurements_by_timerange not fully implemented for VictoriaMetrics") return [] @@ -693,26 +767,27 @@ class VictoriaMetricsAdapter(DatabaseAdapter): logging.warning("get_measurements_for_date not fully implemented for VictoriaMetrics") return [] + # Factory function to create appropriate adapter def create_database_adapter(db_type: str, **kwargs) -> DatabaseAdapter: """ Factory function to create database adapter - + Args: db_type: 'influxdb', 'mysql', 'postgresql', 'sqlite', or 'victoriametrics' **kwargs: Database-specific connection parameters """ db_type = db_type.lower() - - if db_type == 'influxdb': + + if db_type == "influxdb": return InfluxDBAdapter(**kwargs) - elif db_type == 'mysql': - return SQLAdapter(db_type='mysql', **kwargs) - elif db_type == 'postgresql': - return SQLAdapter(db_type='postgresql', **kwargs) - elif db_type == 'sqlite': - return SQLAdapter(db_type='sqlite', **kwargs) - elif db_type == 'victoriametrics': + elif db_type == "mysql": + return SQLAdapter(db_type="mysql", **kwargs) + elif db_type == "postgresql": + return SQLAdapter(db_type="postgresql", **kwargs) + elif db_type == "sqlite": + return SQLAdapter(db_type="sqlite", **kwargs) + elif db_type == "victoriametrics": return VictoriaMetricsAdapter(**kwargs) else: raise ValueError(f"Unsupported database type: {db_type}") diff --git a/src/web_api.py b/src/web_api.py index add641d..6944112 100644 --- a/src/web_api.py +++ b/src/web_api.py @@ -4,26 +4,24 @@ FastAPI web interface for water monitoring system """ import asyncio -import threading -from datetime import datetime, timedelta -from typing import List, Dict, Any, Optional from contextlib import asynccontextmanager +from datetime import datetime, timedelta +from typing import Any, Dict, List, Optional -from fastapi import FastAPI, HTTPException, BackgroundTasks, Depends -from fastapi.responses import HTMLResponse, JSONResponse -from fastapi.staticfiles import StaticFiles +from fastapi import BackgroundTasks, FastAPI, HTTPException from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import HTMLResponse from pydantic import BaseModel, Field -from .water_scraper_v3 import EnhancedWaterMonitorScraper from .config import Config -from .models import WaterMeasurement, StationInfo, ScrapingResult, StationCreateRequest, StationUpdateRequest, StationStatus -from .health_check import HealthCheckManager, DatabaseHealthCheck, APIHealthCheck, MemoryHealthCheck +from .health_check import APIHealthCheck, DatabaseHealthCheck, HealthCheckManager, MemoryHealthCheck +from .logging_config import get_logger, setup_logging from .metrics import get_metrics_collector, increment_counter, set_gauge -from .logging_config import setup_logging, get_logger +from .water_scraper_v3 import EnhancedWaterMonitorScraper logger = get_logger(__name__) + # Pydantic models for API responses class StationResponse(BaseModel): station_id: int @@ -35,6 +33,7 @@ class StationResponse(BaseModel): geohash: Optional[str] = None status: str = "active" + class StationCreateModel(BaseModel): station_code: str = Field(..., description="Station code (e.g., P.1, P.20)") thai_name: str = Field(..., description="Thai name of the station") @@ -44,6 +43,7 @@ class StationCreateModel(BaseModel): geohash: Optional[str] = Field(None, description="Geohash for the location") status: str = Field("active", description="Station status") + class StationUpdateModel(BaseModel): thai_name: Optional[str] = Field(None, description="Thai name of the station") english_name: Optional[str] = Field(None, description="English name of the station") @@ -52,6 +52,7 @@ class StationUpdateModel(BaseModel): geohash: Optional[str] = Field(None, description="Geohash for the location") status: Optional[str] = Field(None, description="Station status") + class MeasurementResponse(BaseModel): timestamp: datetime station_code: str @@ -62,16 +63,19 @@ class MeasurementResponse(BaseModel): discharge_percent: Optional[float] = None status: str = "active" + class HealthResponse(BaseModel): overall_status: str timestamp: str checks: Dict[str, Dict[str, Any]] + class MetricsResponse(BaseModel): counters: Dict[str, float] gauges: Dict[str, float] histograms: Dict[str, Dict[str, float]] + class ScrapingStatusResponse(BaseModel): is_running: bool last_run: Optional[datetime] = None @@ -80,6 +84,7 @@ class ScrapingStatusResponse(BaseModel): successful_runs: int = 0 failed_runs: int = 0 + # Global application state app_state = { "scraper": None, @@ -91,16 +96,17 @@ app_state = { "successful_runs": 0, "failed_runs": 0, "last_run": None, - "next_run": None - } + "next_run": None, + }, } + @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan manager""" # Startup logger.info("Starting Water Monitor API...") - + # Initialize configuration try: Config.validate_config() @@ -108,74 +114,83 @@ async def lifespan(app: FastAPI): except Exception as e: logger.error(f"Configuration validation failed: {e}") raise - + # Initialize scraper db_config = Config.get_database_config() app_state["scraper"] = EnhancedWaterMonitorScraper(db_config) - + # Initialize health checks health_manager = HealthCheckManager() health_manager.add_check(DatabaseHealthCheck(app_state["scraper"].db_adapter)) health_manager.add_check(APIHealthCheck(Config.API_URL, app_state["scraper"].session)) health_manager.add_check(MemoryHealthCheck(max_memory_mb=1000)) app_state["health_manager"] = health_manager - + # Start background scraping task app_state["scraping_task"] = asyncio.create_task(background_scraping_task()) - + logger.info("Water Monitor API started successfully") - + yield - + # Shutdown logger.info("Shutting down Water Monitor API...") - + if app_state["scraping_task"]: app_state["scraping_task"].cancel() try: await app_state["scraping_task"] except asyncio.CancelledError: pass - + logger.info("Water Monitor API shutdown complete") + # Create FastAPI app app = FastAPI( title="Northern Thailand Ping River Monitor API", description="Real-time water level monitoring system for Northern Thailand's Ping River Basin stations", version="3.1.3", - lifespan=lifespan + lifespan=lifespan, ) -# Add CORS middleware +# Add CORS middleware. +# Origins come from CORS_ALLOW_ORIGINS (comma-separated). When none are configured +# we fall back to a wildcard WITHOUT credentials (a safe, spec-valid combination); +# credentials are only enabled when explicit origins are provided. +_cors_origins = Config.CORS_ALLOW_ORIGINS or ["*"] +_cors_allow_credentials = bool(Config.CORS_ALLOW_ORIGINS) app.add_middleware( CORSMiddleware, - allow_origins=["*"], # Configure appropriately for production - allow_credentials=True, + allow_origins=_cors_origins, + allow_credentials=_cors_allow_credentials, allow_methods=["*"], allow_headers=["*"], ) + async def background_scraping_task(): """Background task for periodic data scraping""" while True: try: if not app_state["is_scraping"]: app_state["is_scraping"] = True - + # Run scraping cycle scraper = app_state["scraper"] if scraper: logger.info("Starting background scraping cycle") start_time = datetime.now() - + try: - result = scraper.run_scraping_cycle() - + # 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) + # 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") @@ -184,24 +199,24 @@ async def background_scraping_task(): 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}") - + 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) - + # Wait for next cycle await asyncio.sleep(interval_seconds) - + except asyncio.CancelledError: logger.info("Background scraping task cancelled") break @@ -209,8 +224,10 @@ async def background_scraping_task(): logger.error(f"Error in background scraping task: {e}") await asyncio.sleep(60) # Wait a minute before retrying + # API Routes + @app.get("/", response_class=HTMLResponse) async def root(): """Root endpoint with basic dashboard""" @@ -235,14 +252,14 @@ async def root():

🏔️ Northern Thailand Ping River Monitor API

Real-time water level monitoring system for the Ping River Basin in Northern Thailand

- +

📊 Quick Status

API is running and monitoring 16 water stations along the Ping River

Coverage: From Chiang Dao to Nakhon Sawan

Data collection interval: Every hour

- +

🔗 API Endpoints

GET /health - System health status
@@ -256,7 +273,7 @@ async def root():
GET /scraping/status - Scraping status
GET /docs - Interactive API documentation
- +

📈 Monitoring

• Grafana dashboards available for data visualization

@@ -268,80 +285,86 @@ async def root(): """ return HTMLResponse(content=html_content) + @app.get("/health", response_model=HealthResponse) async def get_health(): """Get system health status""" increment_counter("api_requests", labels={"endpoint": "health"}) - + health_manager = app_state["health_manager"] if not health_manager: raise HTTPException(status_code=503, detail="Health manager not initialized") - - # Run health checks - results = health_manager.run_all_checks() + + # Run health checks (populates state read by get_health_summary) + health_manager.run_all_checks() summary = health_manager.get_health_summary() - + return HealthResponse(**summary) + @app.get("/metrics", response_model=MetricsResponse) async def get_metrics(): """Get application metrics""" increment_counter("api_requests", labels={"endpoint": "metrics"}) - + metrics_collector = get_metrics_collector() metrics = metrics_collector.get_all_metrics() - + return MetricsResponse(**metrics) + @app.get("/stations", response_model=List[StationResponse]) async def get_stations(): """Get list of all monitoring stations""" increment_counter("api_requests", labels={"endpoint": "stations"}) - + scraper = app_state["scraper"] if not scraper: raise HTTPException(status_code=503, detail="Scraper not initialized") - + stations = [] for station_id, station_info in scraper.station_mapping.items(): - stations.append(StationResponse( - station_id=int(station_id), - station_code=station_info["code"], - thai_name=station_info["thai_name"], - english_name=station_info["english_name"], - latitude=station_info.get("latitude"), - longitude=station_info.get("longitude"), - status="active" - )) - + stations.append( + StationResponse( + station_id=int(station_id), + station_code=station_info["code"], + thai_name=station_info["thai_name"], + english_name=station_info["english_name"], + latitude=station_info.get("latitude"), + longitude=station_info.get("longitude"), + status="active", + ) + ) + return stations + @app.post("/stations", response_model=StationResponse) async def create_station(station: StationCreateModel): """Create a new monitoring station""" increment_counter("api_requests", labels={"endpoint": "create_station"}) - + scraper = app_state["scraper"] if not scraper: raise HTTPException(status_code=503, detail="Scraper not initialized") - + try: # Find next available station ID 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)] = { - 'code': station.station_code, - 'thai_name': station.thai_name, - 'english_name': station.english_name, - 'latitude': station.latitude, - 'longitude': station.longitude, - 'geohash': station.geohash + "code": station.station_code, + "thai_name": station.thai_name, + "english_name": station.english_name, + "latitude": station.latitude, + "longitude": station.longitude, + "geohash": station.geohash, } - + logger.info(f"Created new station: {station.station_code} ({station.english_name})") - + return StationResponse( station_id=new_station_id, station_code=station.station_code, @@ -350,271 +373,274 @@ async def create_station(station: StationCreateModel): latitude=station.latitude, longitude=station.longitude, geohash=station.geohash, - status=station.status + status=station.status, ) - + except Exception as e: logger.error(f"Error creating station: {e}") raise HTTPException(status_code=500, detail=str(e)) + @app.put("/stations/{station_id}", response_model=StationResponse) async def update_station(station_id: int, updates: StationUpdateModel): """Update an existing monitoring station""" increment_counter("api_requests", labels={"endpoint": "update_station"}) - + scraper = app_state["scraper"] if not scraper: raise HTTPException(status_code=503, detail="Scraper not initialized") - + station_key = str(station_id) if station_key not in scraper.station_mapping: raise HTTPException(status_code=404, detail="Station not found") - + try: station_info = scraper.station_mapping[station_key] - + # Update fields if provided if updates.thai_name is not None: - station_info['thai_name'] = updates.thai_name + station_info["thai_name"] = updates.thai_name if updates.english_name is not None: - station_info['english_name'] = updates.english_name + station_info["english_name"] = updates.english_name if updates.latitude is not None: - station_info['latitude'] = updates.latitude + station_info["latitude"] = updates.latitude if updates.longitude is not None: - station_info['longitude'] = updates.longitude + station_info["longitude"] = updates.longitude if updates.geohash is not None: - station_info['geohash'] = updates.geohash - + station_info["geohash"] = updates.geohash + logger.info(f"Updated station {station_id}: {station_info['code']}") - + return StationResponse( station_id=station_id, - station_code=station_info['code'], - thai_name=station_info['thai_name'], - english_name=station_info['english_name'], - latitude=station_info.get('latitude'), - longitude=station_info.get('longitude'), - geohash=station_info.get('geohash'), - status=updates.status or "active" + station_code=station_info["code"], + thai_name=station_info["thai_name"], + english_name=station_info["english_name"], + latitude=station_info.get("latitude"), + longitude=station_info.get("longitude"), + geohash=station_info.get("geohash"), + status=updates.status or "active", ) - + except Exception as e: logger.error(f"Error updating station {station_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) + @app.delete("/stations/{station_id}") async def delete_station(station_id: int): """Delete a monitoring station""" increment_counter("api_requests", labels={"endpoint": "delete_station"}) - + scraper = app_state["scraper"] if not scraper: raise HTTPException(status_code=503, detail="Scraper not initialized") - + station_key = str(station_id) if station_key not in scraper.station_mapping: raise HTTPException(status_code=404, detail="Station not found") - + try: station_info = scraper.station_mapping.pop(station_key) logger.info(f"Deleted station {station_id}: {station_info['code']}") - + return {"message": f"Station {station_info['code']} deleted successfully"} - + except Exception as e: logger.error(f"Error deleting station {station_id}: {e}") raise HTTPException(status_code=500, detail=str(e)) + @app.get("/stations/{station_id}", response_model=StationResponse) async def get_station(station_id: int): """Get details of a specific monitoring station""" increment_counter("api_requests", labels={"endpoint": "get_station"}) - + scraper = app_state["scraper"] if not scraper: raise HTTPException(status_code=503, detail="Scraper not initialized") - + station_key = str(station_id) if station_key not in scraper.station_mapping: raise HTTPException(status_code=404, detail="Station not found") - + station_info = scraper.station_mapping[station_key] - + return StationResponse( station_id=station_id, - station_code=station_info['code'], - thai_name=station_info['thai_name'], - english_name=station_info['english_name'], - latitude=station_info.get('latitude'), - longitude=station_info.get('longitude'), - geohash=station_info.get('geohash'), - status="active" + station_code=station_info["code"], + thai_name=station_info["thai_name"], + english_name=station_info["english_name"], + latitude=station_info.get("latitude"), + longitude=station_info.get("longitude"), + geohash=station_info.get("geohash"), + status="active", ) + @app.get("/measurements/latest", response_model=List[MeasurementResponse]) async def get_latest_measurements(limit: int = 100): """Get latest measurements from all stations""" increment_counter("api_requests", labels={"endpoint": "measurements_latest"}) - + scraper = app_state["scraper"] if not scraper or not scraper.db_adapter: raise HTTPException(status_code=503, detail="Database not available") - + try: measurements = scraper.get_latest_data(limit=limit) - + response = [] for measurement in measurements: - response.append(MeasurementResponse( - timestamp=measurement["timestamp"], - station_code=measurement["station_code"], - station_name_en=measurement["station_name_en"], - station_name_th=measurement["station_name_th"], - water_level=measurement["water_level"], - discharge=measurement["discharge"], - discharge_percent=measurement.get("discharge_percent"), - status=measurement.get("status", "active") - )) - + response.append( + MeasurementResponse( + timestamp=measurement["timestamp"], + station_code=measurement["station_code"], + station_name_en=measurement["station_name_en"], + station_name_th=measurement["station_name_th"], + water_level=measurement["water_level"], + discharge=measurement["discharge"], + discharge_percent=measurement.get("discharge_percent"), + status=measurement.get("status", "active"), + ) + ) + return response - + except Exception as e: logger.error(f"Error fetching latest measurements: {e}") raise HTTPException(status_code=500, detail=str(e)) + @app.get("/measurements/station/{station_code}", response_model=List[MeasurementResponse]) -async def get_station_measurements( - station_code: str, - hours: int = 24, - limit: int = 1000 -): +async def get_station_measurements(station_code: str, hours: int = 24, limit: int = 1000): """Get measurements for a specific station""" increment_counter("api_requests", labels={"endpoint": "measurements_station"}) - + scraper = app_state["scraper"] if not scraper or not scraper.db_adapter: raise HTTPException(status_code=503, detail="Database not available") - + try: # Get measurements for the specified time range end_time = datetime.now() start_time = end_time - timedelta(hours=hours) - + measurements = scraper.db_adapter.get_measurements_by_timerange( start_time, end_time, station_codes=[station_code] ) - + # Limit results measurements = measurements[:limit] - + response = [] for measurement in measurements: - response.append(MeasurementResponse( - timestamp=measurement["timestamp"], - station_code=measurement["station_code"], - station_name_en=measurement["station_name_en"], - station_name_th=measurement["station_name_th"], - water_level=measurement["water_level"], - discharge=measurement["discharge"], - discharge_percent=measurement.get("discharge_percent"), - status=measurement.get("status", "active") - )) - + response.append( + MeasurementResponse( + timestamp=measurement["timestamp"], + station_code=measurement["station_code"], + station_name_en=measurement["station_name_en"], + station_name_th=measurement["station_name_th"], + water_level=measurement["water_level"], + discharge=measurement["discharge"], + discharge_percent=measurement.get("discharge_percent"), + status=measurement.get("status", "active"), + ) + ) + return response - + except Exception as e: logger.error(f"Error fetching station measurements: {e}") raise HTTPException(status_code=500, detail=str(e)) + @app.post("/scrape/trigger") async def trigger_scraping(background_tasks: BackgroundTasks): """Trigger manual data scraping""" increment_counter("api_requests", labels={"endpoint": "scrape_trigger"}) - + if app_state["is_scraping"]: raise HTTPException(status_code=409, detail="Scraping already in progress") - + scraper = app_state["scraper"] if not scraper: raise HTTPException(status_code=503, detail="Scraper not initialized") - + def run_scraping(): """Background task to run scraping""" try: app_state["is_scraping"] = True logger.info("Manual scraping triggered via API") - + 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") - + except Exception as e: app_state["scraping_stats"]["failed_runs"] += 1 increment_counter("manual_scraping_failed") logger.error(f"Manual scraping failed: {e}") finally: app_state["is_scraping"] = False - + background_tasks.add_task(run_scraping) - + return {"message": "Scraping triggered", "status": "started"} + @app.get("/scraping/status", response_model=ScrapingStatusResponse) async def get_scraping_status(): """Get current scraping status""" increment_counter("api_requests", labels={"endpoint": "scraping_status"}) - + stats = app_state["scraping_stats"] - + 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"] + failed_runs=stats["failed_runs"], ) + @app.get("/config") async def get_config(): """Get current configuration (sensitive data masked)""" increment_counter("api_requests", labels={"endpoint": "config"}) - + config = Config.get_all_settings() - + # Mask sensitive information for key in config: - if 'password' in key.lower() or 'secret' in key.lower(): + if "password" in key.lower() or "secret" in key.lower(): if config[key]: - config[key] = '*' * 8 - + config[key] = "*" * 8 + return config + if __name__ == "__main__": import uvicorn - + # Setup logging setup_logging( log_level=Config.LOG_LEVEL, log_file=Config.LOG_FILE, enable_console=True, - enable_colors=True + enable_colors=True, ) - + # Run the API server - uvicorn.run( - "web_api:app", - host="0.0.0.0", - port=8000, - reload=False, - log_config=None # Use our custom logging - ) \ No newline at end of file + uvicorn.run("web_api:app", host="0.0.0.0", port=8000, reload=False, log_config=None) # Use our custom logging